package net.bwie.realtime.function;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SourceSplitFunction extends ProcessFunction<String, String> {

    private final OutputTag<String> carTag;
    private final OutputTag<String> carLogTag;
    private final OutputTag<String> owerTag;
    private final OutputTag<String> entranceGuardLogTag;
    private final OutputTag<String> roomTag;

    public SourceSplitFunction(OutputTag<String> carTag, OutputTag<String> carLogTag,
                               OutputTag<String> owerTag, OutputTag<String> entranceGuardLogTag,
                               OutputTag<String> roomTag) {
        this.carTag = carTag;
        this.carLogTag = carLogTag;
        this.owerTag = owerTag;
        this.entranceGuardLogTag = entranceGuardLogTag;
        this.roomTag = roomTag;
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        try {

            JSONObject jsonObject = JSON.parseObject(value);
            if (jsonObject == null) {
                return;
            }

            JSONObject source = jsonObject.getJSONObject("source");
            JSONObject after = jsonObject.getJSONObject("after");

            // 检查必要的字段是否存在
            if (source == null || after == null) {
                return;
            }

            String tableName = source.getString("table");

            switch (tableName) {
                case "cars_Log":
//                    JSONObject after1 = processCarLog(after);
                    ctx.output(carLogTag, after.toJSONString());
                    break;

                case "cars":
                    ctx.output(carTag, after.toJSONString());
                    break;

                case "entrance_guard_log":
                    ctx.output(entranceGuardLogTag, after.toJSONString());
                    break;

                case "ower":
                    processOwer(after);
                    ctx.output(owerTag, after.toJSONString());
                    break;

                case "room":
                    ctx.output(roomTag, after.toJSONString());
                    break;

                default:
                    // 可以记录未知表名的日志
                    break;
            }

        } catch (Exception e) {
            // 建议添加日志记录异常
            System.err.println("Error processing element: " + value);
            e.printStackTrace();
        }
    }

    /**
     * 处理 cars_Log 表的 ctype 字段转换
     */
    private JSONObject processCarLog(JSONObject after) {
        Integer ctype = after.getInteger("ctype");
        if (ctype != null) {
            if (ctype == 1) {
                after.put("ctype", "驶入");
            } else {
                after.put("ctype", "驶出");
            }
        }

        return after;
    }

    /**
     * 处理 ower 表的 openMode 字段转换
     */
    private void processOwer(JSONObject after) {
        Integer openMode = after.getInteger("openMode");
        if (openMode != null) {
            switch (openMode) {
                case 1:
                    after.put("openMode", "刷卡");
                    break;
                case 2:
                    after.put("openMode", "刷APP");
                    break;
                case 3:
                    after.put("openMode", "人脸识别");
                    break;
                case 4:
                    after.put("openMode", "远程");
                    break;
                default:
                    // 可以保留原值或设置为未知
                    after.put("openMode", "未知(" + openMode + ")");
                    break;
            }
        }
    }
}